package com.dc.dbus.transport.rabbitmq;

import java.util.HashMap;

import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;

import com.dc.dbus.transport.commons.ServerSideReceiver;
import com.dc.dbus.transport.commons.ServerSideTransport;

public class RabbitMqServerSideTransport implements ServerSideTransport {
	
	protected final RabbitTemplate template;
	
	protected final RabbitMqManager rabbitMqManager;
	
	protected final NoExceptionEventListener eventListener;
	
	protected final String serverQueueName;
	
	public RabbitMqServerSideTransport(
			RabbitMqManager rabbitMqManager, EventListener eventListener,
			String serverQueueName, Integer maxLength, Integer messageTtl, boolean durable, boolean exclusive, boolean autoDelete) {
		super();
		this.template = rabbitMqManager.getTemplate();
		this.rabbitMqManager = rabbitMqManager;
		this.eventListener = new NoExceptionEventListener(eventListener);
		this.serverQueueName = serverQueueName;
		
		HashMap<String, Object> params = new HashMap<String, Object>();
		if (maxLength != null) {
			params.put("x-max-length", maxLength);
		}
		if (messageTtl != null) {
			params.put("x-message-ttl", messageTtl);
		}
		
		Queue serverQueue = new Queue(serverQueueName, durable, exclusive, autoDelete, params);
		rabbitMqManager.getAdmin().declareQueue(serverQueue);
	}
	
	@Override
	public void startServerSideReceiver(final ServerSideReceiver serverSideReceiver) {
		
		MessageListenerAdapter listenerAdapter = new MessageListenerAdapter(new Object() {
			@SuppressWarnings("unused")
			public void handleMessage(Object param) {
				
				try {
					
					if (param instanceof Msg) {
						Msg msg = (Msg) param;
						serverSideReceiver.handlerMsg(msg.source, msg.msgId, msg.cmd, msg.param);
					} else {
						eventListener.exception("unknow msg type from client, msg type:" + param.getClass().toString(), 
								new RuntimeException());
					}
					
				} catch (Exception e) {
					eventListener.exception("RabbitMqClientSideTansport.handleMsg", e);
				}
				
			}
		});
		
        SimpleMessageListenerContainer listener = new SimpleMessageListenerContainer(rabbitMqManager.getCf());
        listener.setConcurrentConsumers(1);
        listener.setMessageListener(listenerAdapter);
        listener.setQueueNames(serverQueueName);
        listener.setAcknowledgeMode(AcknowledgeMode.NONE);
        listener.setChannelTransacted(false);
        listener.start();
	}
	
	@Override
	public void sendBack(String source, long replyForMsgId, int replyForCmd, int resultCode, Object returnObj) {
		
    	MsgBack msgBack = new MsgBack(replyForMsgId, replyForCmd, resultCode, returnObj);
		template.convertAndSend("", source, msgBack);
	}

	@Override
	public void close() {
		// TODO 未实现关闭
	}
	
	
	
}
